﻿using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service.Template;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Exceptions;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Mutation;
using Me.Prettyprint.Hector.Api.Query;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Service
{
    /// <summary>
    /// HColumnFamilyImpl encapsulates the results of a query and has methods
    /// that execute queries against the server.
    /// </summary>
    /// <typeparam name="K"></typeparam>
    /// <typeparam name="N"></typeparam>
    public class HColumnFamilyImpl<K, N> : HColumnFamily<K, N>
    {
        private Logger queryLogger = new Logger(typeof(HColumnFamilyImpl<K,N>));

        public HColumnFamilyImpl(Keyspace keyspace, String columnFamilyName, Serializer<K> keySerializer, Serializer<N> columnNameSerializer)
        {
            Keyspace = (ExecutingKeyspace)keyspace;
            ColumnFamilyName = columnFamilyName;
            Keys = new List<K>();
            KeySerializer = keySerializer;
            ColumnNameSerializer = columnNameSerializer;
            ActiveSlicePredicate = new HSlicePredicate<N>(columnNameSerializer);
            ColumnParent = new ColumnParent(columnFamilyName);
            ExceptionsTranslator = new ExceptionsTranslatorImpl();
            ConsistencyLevelPolicy = new ConfigurableConsistencyLevel();
        }
        public long ExecutionTimeMicro
        {
            get;
            set;
        }
        public long ExecutionTimeNano
        {
            get;
            set;
        }
        protected ExecutingKeyspace Keyspace
        {
            get;
            set;
        }
        protected string ColumnFamilyName
        {
            get;
            set;
        }
        protected List<K> Keys
        {
            get;
            set;
        }
        private HSlicePredicate<N> ActiveSlicePredicate
        {
            get;
            set;
        }
        private HSlicePredicate<N> LastAppliedPredicate
        {
            get;
            set;
        }
        private Serializer<K> KeySerializer
        {
            get;
            set;
        }
        private Serializer<N> ColumnNameSerializer
        {
            get;
            set;
        }
        private ConfigurableConsistencyLevel ConsistencyLevelPolicy
        {
            get;
            set;
        }
        private Dictionary<N, HColumn<N, byte[]>> Columns
        {
            get;
            set;
        }
        private ColumnParent ColumnParent
        {
            get;
            set;
        }
        private ExceptionsTranslator ExceptionsTranslator
        {
            get;
            set;
        }
        private bool HasValues
        {
            get;
            set;
        }
        private List<N> ColumnNames
        {
            get;
            set;
        }
        private int RowIndex
        {
            get;
            set;
        }
        private Dictionary<byte[], List<ColumnOrSuperColumn>> Rows
        {
            get;
            set;
        }
        private CassandraHost LastHostUsed
        {
            get;
            set;
        }
        public CassandraHost HostUsed
        {
            get;
            set;
        }
        private long LastExecutionTime
        {
            get;
            set;
        }
        public HColumnFamily<K, N> AddKey(K key)
        {
            Keys.Add(key);
            return this;
        }
        public HColumnFamily<K, N> AddKeys(List<K> keys)
        {
            foreach (K key in keys)
            {
                Keys.Add(key);
            }
            return this;
        }
        public HColumnFamily<K, N> RemoveKeys()
        {
            Keys.Clear();
            return this;
        }
        public HColumnFamily<K, N> SetCount(int count)
        {
            ActiveSlicePredicate.SetCount(count);
            return this;
        }
        public HColumnFamily<K, N> SetFinish(N name)
        {
            ActiveSlicePredicate.SetEndOn(name);
            return this;
        }
        public HColumnFamily<K, N> SetReversed(bool reversed)
        {
            ActiveSlicePredicate.SetReversed(reversed);
            return this;
        }
        public HColumnFamily<K, N> SetStart(N name)
        {
            ActiveSlicePredicate.SetStartOn(name);
            return this;
        }
        public HColumnFamily<K, N> AddColumnName(N columnName)
        {
            ActiveSlicePredicate.AddColumnName(columnName);
            return this;
        }
        public HColumnFamily<K, N> SetColumnNames(List<N> columnNames)
        {
            ActiveSlicePredicate.SetColumnNames(columnNames);
            return null;
        }
        /// <summary>
        /// Note this method just returns null, since Hector returns null.
        /// Not sure exactly why it does that.
        /// </summary>
        /// <param name="name"></param>
        /// <returns></returns>
        public HColumn<N,object> GetColumn(N name)
        {
            return null;
        }
        public List<HColumn<N, byte[]>> GetColumns()
        {
            if (Columns == null)
            {
                Columns = new Dictionary<N, HColumn<N, byte[]>>();
            }
            if (!HasValues)
            {
                DoExecuteSlice();
            }
            return new List<HColumn<N,byte[]>>(Columns.Values);
        }
        public HColumnFamily<K, N> Clear()
        {
            foreach (HColumn<N,ByteBuffer> col in Columns.Values)
            {
                // could probably do something fancier hear calling clear() on the underlying BB
                col.Clear();
            }
            HasValues = false;
            return this;
        }
        public void Dispose()
        {
            Clear();
        }
        public DateTime GetDate(N name)
        {
            return ExtractColumnValue(name, DateSerializer.Instance);
        }
        public double GetDouble(N name)
        {
            return ExtractColumnValue(name, DoubleSerializer.Instance);
        }
        public int GetInt(N name)
        {
            return ExtractColumnValue(name, IntegerSerializer.Instance);
        }
        public long GetLong(N name)
        {
            return ExtractColumnValue(name, LongSerializer.Instance);
        }
        public string GetString(N name)
        {
            return ExtractColumnValue(name, StringSerializer.Instance);
        }
        public Guid GetUUID(N name)
        {
            return ExtractColumnValue(name, UUIDSerializer.Instance);
        }
        public HColumnFamily<K, N> Next()
        {
            if ( !HasNext() )
            {
                throw new SystemException("No more rows left on this HColumnFamily");
            }
            RowIndex++;
            K key = Keys[RowIndex];
            ApplyToRow(key, Rows[KeySerializer.ToBytes(key)]);
            return this;
        }
        public void Reset()
        {
            RowIndex = 0;
        }
        public HColumnFamily<K, N> Current
        {
            get
            {
                K key = Keys[RowIndex];
                ApplyToRow(key, Rows[KeySerializer.ToBytes(key)]);
                return this;
            }
        }
        public bool MoveNext()
        {
            if (HasNext())
            {
                Next();
                return true;
            }
            else
            {
                return false;
            }
        }
        public bool HasNext()
        {
            return RowIndex < Rows.Count - 1 ;
        }
        public void Remove()
        {
            K key = Keys[RowIndex];
            Keys.Remove(key);
            Rows.Remove(KeySerializer.ToBytes(key));
        }
        public V GetValue<V>(N name, Serializer<V> valueSerializer)
        {
            return ExtractColumnValue(name, valueSerializer);
        }
        public HColumnFamily<K, N> SetReadConsistencyLevel(HConsistencyLevel readLevel)
        {
            ConsistencyLevelPolicy.setDefaultReadConsistencyLevel(readLevel);
            return this;
        }
        public HColumnFamily<K, N> SetWriteConsistencyLevel(HConsistencyLevel writeLevel)
        {
            ConsistencyLevelPolicy.setDefaultWriteConsistencyLevel(writeLevel);
            return this;
        }
        public long GetExecutionTimeMicro()
        {
            return LastExecutionTime / 1000;
        }
        public CassandraHost GetHostUsed()
        {
            return LastHostUsed;
        }
        private V ExtractColumnValue<V>(N columnName, Serializer<V> valueSerializer)
        {
            MaybeExecuteSlice(columnName);
            return Columns[columnName] != null && Columns[columnName].Value != null ? valueSerializer.FromBytes(Columns[columnName].Value) : default(V);
        }
        private void MaybeExecuteSlice(N columnName)
        {
            if (ColumnNames == null)
            {
                ColumnNames = new List<N>();
            }
            if (Columns == null)
            {
                Columns = new Dictionary<N, HColumn<N,byte[]>>();
            }
            if (Columns[columnName] == null)
            {
                ColumnNames.Add(columnName);
                ActiveSlicePredicate.SetColumnNames(ColumnNames);
                if (Keys.Count == 1 )
                {
                    DoExecuteSlice();
                }
                else
                {
                    DoExecuteMultigetSlice();
                }      
            }
        }
        public void ApplyToRow(K key, List<ColumnOrSuperColumn> cosclist)
        {
            HColumn<N, byte[]> column;
            N colName;
            for (IEnumerator<ColumnOrSuperColumn> iterator = cosclist.GetEnumerator(); iterator.MoveNext();)
            {
                ColumnOrSuperColumn cosc = iterator.Current;
                colName = ColumnNameSerializer.FromBytes(cosc.Column.Name);
                column = Columns[colName];
                if ( column == null )
                {
                    column = new HColumnImpl<N, byte[]>(cosc.Column, ColumnNameSerializer, BytesArraySerializer.Instance);
                } 
                else
                {
                    ((HColumnImpl<N, byte[]>)column).Apply(cosc.Column);
                }
                Columns[colName] = column; 
            }
        }
        public void ApplyResultStatus(long execTime, CassandraHost cassandraHost) 
        {
            LastExecutionTime = execTime;
            LastHostUsed = cassandraHost;
        }
        public void DoExecuteSlice()
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Column> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<ColumnOrSuperColumn> cosclist = cassandra.get_slice(KeySerializer.ToBytes(Keys[0]), ColumnParent,
                        ActiveSlicePredicate.ToThrift(), 
                        ThriftConverter.consistencyLevel(ConsistencyLevelPolicy.Get(OperationType.READ)));
                    ApplyToRow(Keys[0], cosclist);
                    HasValues = true;
                    return null;
                }
                catch (Exception e)
                {
                    throw ExceptionsTranslator.translate(e);
                }
            };
            Operation<Column> op = new OperationBuilder<Column>(x);
            op.OperationType = OperationType.READ;
            op.KeyspaceName = Keyspace.KeyspaceName;
            Keyspace.DoExecuteOperation(op);
            ApplyResultStatus(op.ExecTime, op.CassandraHost);
        }
        public void DoExecuteMultigetSlice()
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, Column> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    Rows = cassandra.multiget_slice(KeySerializer.ToBytesList(Keys), ColumnParent,
                        ActiveSlicePredicate.ToThrift(), 
                        ThriftConverter.consistencyLevel(ConsistencyLevelPolicy.Get(OperationType.READ)));
                    HasValues = true;
                    return null;
                }
                catch (Exception e)
                {
                    throw ExceptionsTranslator.translate(e);
                }
            };
            Operation<Column> op = new OperationBuilder<Column>(x);
            op.OperationType = OperationType.READ;
            op.KeyspaceName = Keyspace.KeyspaceName;
            Keyspace.DoExecuteOperation(op);
            ApplyResultStatus(op.ExecTime, op.CassandraHost);
            ApplyToRow(Keys[0], Rows[KeySerializer.ToBytes(Keys[0])]);
        }
        public long GetExecutionTimeNano()
        {
            return LastExecutionTime;
        }
    }
}
